package net.bwie.realtime.jtp.log.job;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/22 19:29:46
 * @version: 1.0
 */
public class jtplogEtlDwdjob {
    public static void main(String[] args) throws Exception {
        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> dwd_traffic_page_log = KafkaUtil.consumerKafka(env, "dwd_traffic_page_log");

        DataStream<String> dwd_traffic_start_log = KafkaUtil.consumerKafka(env, "dwd_traffic-start-log"); // 假设有另一个topic

        // 处理页面日志
        JtpLogClickHouse.savePageLog(dwd_traffic_page_log);

        // 处理启动日志
        JtpLogClickHouse.saveStartLog(dwd_traffic_start_log);

        env.execute("jtplogEtlDwdjob");
    }
}
